package com.hubu.thread.threadpool;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ThreadFactory;
/**
 * 线程池
 */
public class ThreadPool {

    private ThreadFactory threadFactory;
    private Set<Worker> workerSet=new HashSet<>();
    public  BlockingQueue<Runnable> taskQueue=null;
    /**
     * 核心线程数
     */
    private int coreSize;
    /**
     * 最大线程数
     */
    private int maxTaskSize;
    private int maxSize;

    /**
     *
     * 拒绝策略
     */
    private RejectPolicy rejectPolicy;
    public ThreadPool(
            int coreSize,
            int maxSize,
            int maxTaskSize,
            ThreadFactory threadFactory,
            RejectPolicy rejectPolicy){

        this.maxTaskSize=maxTaskSize;
        this.coreSize=coreSize;
        this.maxSize=maxSize;
        taskQueue=new BlockingQueue<>(maxTaskSize);
        this.threadFactory=threadFactory;
        this.rejectPolicy=rejectPolicy;
    }
    /**
     *
     * 线程池执行任务
     * @param task
     */
    public void execute(Runnable task) {
        synchronized (workerSet){
            if(workerSet.size()<coreSize) {
                Worker worker=new Worker(task);
                workerSet.add(worker);
                worker.start();
            }
            else{
                if(taskQueue.size()==maxTaskSize){
                    //任务队列已经满了
                    rejectPolicy.reject(task);
                }
                else{
                    taskQueue.offer(task);
                }
            }
        }
    }
    class Worker extends Thread{
        private Runnable task;
        public Worker(Runnable task){
            this.task=task;
        }
        //线程执行函数
        @Override
        public void run() {
            while(task!=null ||(task=taskQueue.poll())!=null){
                try{
                    task.run();
                }catch (Exception e){
                    e.printStackTrace();
                }
                finally {
                    task=null;
                }
            }
            synchronized (workerSet){
                workerSet.remove(this);
            }
        }
    }
    static class BlockingQueue<T> {
        private final Queue<T> queue=new LinkedList<>();
        private int maxSize;
        public BlockingQueue(int maxSize){
            this.maxSize=maxSize;
        }
        public void offer(T data){
            synchronized (queue){
                while(queue.size()==maxSize){
                    try {
                        //自己wait住 释放queue锁 其他线程获取锁
                        queue.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                queue.offer(data);
                synchronized (queue){
                    queue.notifyAll();
                }
            }
        }
        public T poll(){
            synchronized (queue){
                while(queue.isEmpty()){
                    try {
                        queue.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            T value = queue.poll();
            synchronized (queue){
                queue.notifyAll();
            }
            return value;
        }
        public int size(){
            synchronized (queue){
                return queue.size();
            }
        }
    }
}
